package javacourse.mq.mymq.Queue;

import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Create by chenhd on 2021/11/6.
 */
@Component
public class MyQueue {
    //partition个数，这里暂时直接写死
    public static final int PARTITIONS = 3;
    //消息队列，记录每个topicPartition中的消息列表
    Map<TopicPartition, List<String>> messageQueue = new ConcurrentHashMap<>();
    //记录topicPartition当前消费进度
    Map<TopicPartition, Integer> topicPartitionOffsetMap = new ConcurrentHashMap<>();
    //生产者生产消息时，当前topic的消息应发送到哪个partition
    Map<String, Integer> topicPartitionMap = new ConcurrentHashMap<>();
    //记录消费者分配的partition
    Map<TopicPartition, String> partitionConsumerMap = new ConcurrentHashMap<>();

    final String partitionMutex = "";
    final String assignMutex = "";

    Lock messageLock = new ReentrantLock();
    Map<TopicPartition, Condition> conditionMap = new HashMap<>();

    /**
     * 生产者获取下一个partition，当前直接采用轮询
     *
     * @param topic
     * @return
     */
    public Integer getNextPartition(String topic) {
        Integer partition;
        synchronized (partitionMutex) {
            partition = topicPartitionMap.computeIfAbsent(topic, s -> 0);
            topicPartitionMap.put(topic, (partition + 1) % PARTITIONS);
        }
        return partition;
    }

    /**
     * 为consumer分配partition，这里直接按传入的consumers顺序分配
     * 如果consumers多于分区数则一个消费者可能分配到多个partition，如果少于分区数则有的consumer没有partition
     *
     * @param topic
     * @param consumers
     */
    public void assignPartition(String topic, List<String> consumers) {
        synchronized (assignMutex) {
            for (int i = 0; i < PARTITIONS; i++) {
                TopicPartition topicPartition = new TopicPartition(topic, i);
                partitionConsumerMap.put(topicPartition, consumers.get(i % consumers.size()));
            }
        }
    }

    /**
     * 根据topicPartition获取消息列表
     *
     * @param topicPartition
     * @return
     */
    public List<String> getQueue(TopicPartition topicPartition) {
        return messageQueue.computeIfAbsent(topicPartition, topicPartition1 -> {
            conditionMap.put(topicPartition1, messageLock.newCondition());
            return new ArrayList<>();
        });
    }

    /**
     * 获取offset
     *
     * @param topicPartition
     * @return
     */
    public Integer getOffset(TopicPartition topicPartition) {
        return topicPartitionOffsetMap.computeIfAbsent(topicPartition, topicPartition1 -> 0);
    }

    /**
     * 设置offset
     *
     * @param topicPartition
     * @param offset
     * @return
     */
    public Integer setOffset(TopicPartition topicPartition, Integer offset) {
        return topicPartitionOffsetMap.put(topicPartition, offset);
    }


    public Map<TopicPartition, String> getPartitionConsumerMap() {
        return partitionConsumerMap;
    }

    public Map<TopicPartition, Condition> getConditionMap() {
        return conditionMap;
    }

    public Lock getMessageLock() {
        return messageLock;
    }
}
